[Amazon Connect]問い合わせレコードデータモデル(CTR)をS3に保存し、Athenaを使って通話録音データの格納場所を検索する

Amazon Connectには、問い合わせレコードデータモデル(以下、CTR)というデータがあります。
CTRはAmazon Connectで発生するイベントや、属性、キュー、エージェントのやり取りをキャプチャしたものになります。
Amazon Connectから通話録音を再生するときも、このCTRというデータを利用して該当のS3の場所を特定し再生を行います。

CTRはAmazon Connectの仕様として24ヶ月までしか保存できません。
その為、2年以上の音声データを再生したい時などはCTRをAmazon Connectではなく別で管理する必要があります。

今回はCTRをAmazon S3に格納し、Amazon Athenaを利用してCTRからS3に保存されている録音データの場所を特定してみます。

※CTRから見れるデータは下記ドキュメントからご確認ください。

構成図

構成図はこちらです。

Amazon Connectから直接Amazon S3に通話録音データ(wavファイル)を格納します。
それとは別に、Amazon Kinesis Data Streams、Amazon Kinesis Data Firehoseを通ってCTRをAmazon S3の別バケットに格納します。
Amazon Athenaを利用してCTRを保存しているバケットからクエリを行い、S3にある通話録音データの場所を特定します。
今回、KinesisについてAmazon Kinesis Data Streamsを挟まなくてもAmazon Connect→Amazon Kinesis Data FirehoseからS3に保存可能ですが、Amazon Connectは一つのKinesis(Data Streams か Data Firehose)しか設定ができない為、今後他にKinesisを利用することを見越して、Amazon Kinesis Data Streamsを挟んで構築します。
(Amazon Kinesis Data Streamsを設定しておけば、そこから複数Amazon Kinesis Data Firehoseの設定が可能です)

やってみる

CTR用のAmazon S3を作成

コンソールからAmazon S3を開き、バケットを作成します。
バケット名はグローバルで一意の名前を入力し、その他はデフォルトで設定します。

Amazon Kinesisの作成

Kinesis Data Firehoseに流すKinesis Data Streamsを作成します。
検証用のため、プロビジョンドモードでシャード数は1とします。

ステータスがアクティブになったら、下のコンシューマーからAmazon Kinesis Data Firehoseをクリックします。

送信先をAmazon S3に設定し、送信先の設定でS3バケットを先ほど作成したバケットで選択します。
S3バケットプレフィックスで、バケットの中でプレフィックスを指定します。(今回はcontact-trace-recordsとします)
エラーの場合も出力プレフィックスでerrorに入るように指定します。
その他はデフォルトで進めていきます。

Amazon Athenaの設定

Amazon Athenaのコンソールを開き、管理をクリックします。

作成したS3バケットを指定して、保存をクリックします。
クエリ画面で、CTR用のテーブルを作成するために、下記SQLを入力します。
(このコードはSINさんのブログを参照しています。SINさん、有難うございます！)

CREATE EXTERNAL TABLE IF NOT EXISTS `default`.`ctr_table` (
  Agent string,
  AgentConnectionAttempts int,
  Attributes string,
  AWSAccountId string,
  AWSContactTraceRecordFormatVersion string,
  Channel string,
  ConnectedToSystemTimestamp string,
  ContactId string,
  CustomerEndpoint string,
  DisconnectTimestamp string,
  DisconnectReason string,
  InitialContactId string,
  InitiationMethod string,
  InitiationTimestamp string,
  InstanceARN string,
  LastUpdateTimestamp string,
  MediaStreams string,
  NextContactId string,
  PreviousContactId string,
  Queue string,
  Recording string,
  Recordings string,
  SystemEndpoint string,
  TransferCompletedTimestamp string,
  TransferredToEndpoint string
)
PARTITIONED BY (
  year int,
  month int,
  day int,
  hour int
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe' 
WITH SERDEPROPERTIES (
  'serialization.format' = '1'
) LOCATION 's3://${作成したバケット名}/contact-trace-records/'
TBLPROPERTIES (
  'has_encrypted_data'='false',

  'projection.enabled' = 'true',

  'projection.year.type' = 'integer',
  'projection.year.range' = '2021,2100',
  'projection.year.digits' = '4',

  'projection.month.type' = 'integer',
  'projection.month.range' = '1,12',
  'projection.month.digits' = '2',

  'projection.day.type' = 'integer',
  'projection.day.range' = '1,31',
  'projection.day.digits' = '2',
   
  'projection.hour.type' = 'integer',
  'projection.hour.range' = '0,23',
  'projection.hour.digits' = '2',
  
  'storage.location.template' = 's3://${作成したバケット名}/contact-trace-records/${year}/${month}/${day}/${hour}'
);

実行するをクリックし、完了済みになったら、新しくctr_tableが作成できていることを確認します。

Amazon Connectの設定

最後にAmazon Connectの設定を行います。
インスタンスからデータストリーミングをクリックし、作成したKinesisストリームを指定して保存をクリックします。

データストレージで通話記録も設定されていることを確認しましょう。

ここまでできたら構築は完了です。

検証

それでは検証を行います。
まずは登録している電話番号に着信させます。
電話をかけて数分待つと、S3のバケットの中にcontact-trace-recordsのフォルダが作成されます。
year、month、day、hourの順で階層があり、その先にCTRデータが格納されています。

CTRデータを確認できたら、Athenaにいき、下記クエリを実行します。

SELECT 
    ConnectedToSystemTimestamp,
    DisconnectTimestamp,
    Channel,
    ContactId,
    json_extract(Queue, '$.name') AS Queue,
    json_extract(Agent, '$.username') AS AgentName,
    json_extract(Recording, '$.location') AS RecordingLocation,
    json_extract(CustomerEndpoint, '$.address') AS CustomerEndpoint,
    json_extract(SystemEndpoint, '$.address') AS SystemEndpoint
FROM "default"."ctr_table"
WHERE year = 2022 and month=10 and day=5
limit 10

上の内容はCTRのデータをクエリしています。
他に取得したいCTRデータがあれば、SELECT文の中に記載します。
json_extract関数を利用することで、対象の値のみを抜き出すことができます。
WHERE句を利用することで、対象の日時を絞ることができます。
クエリを実行すると、エージェントの情報等とともに、RecordingLocation(録音場所)を特定することができました。

対象のS3バケットを見にいくと、対象の録音データを見つけることができました！

最後に

CTRの24ヶ月制限の仕様において、通話録音が2年以上前は再生できないことが主にあげられます。
そこまで遡って再生しなくてよければ、デフォルトの録音再生機能を利用すれば事足りますが、社内ポリシー等で2年以上の録音再生が必須の場合は今回のようにCTRを別のところに保存するやり方が必要です。
2年以上前の録音データの再生頻度がそこまで高くなければ、都度Athenaでクエリをかけて探す方法でいいかと思います。
逆に頻繁に検索する、フロント画面も欲しい等ある場合は、何度かご紹介しているSINさんの下記ブログをご参照ください。

ではまた！コンサルティング部の洲崎でした。